package com.wykj.business.service;

import java.util.List;


import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.stereotype.Service;

import com.wykj.business.model.Test;

@Service
public class HbaseService {
	
	@Autowired 
    private HbaseTemplate hbaseTemplate;  
    // 获取表中指定行，列簇，列的值  
    public String get(String tableName, String rowName, String familyName,  
            String qualifier) {  
        return hbaseTemplate.get(tableName, rowName, familyName, qualifier,  
                new RowMapper<String>() {  
                    public String mapRow(Result result, int rowNum)  
                            throws Exception {  
                        List<Cell> ceList = result.listCells();  
                        String res = "";  
                        if (ceList != null && ceList.size() > 0) {  
                            for (Cell cell : ceList) {  
                                res = Bytes.toString(cell.getValueArray(),  
                                        cell.getValueOffset(),  
                                        cell.getValueLength());  
                            }  
                        }  
                        return res;  
                    }  
                });  
    }  
    // 浏览表中某些行的信息，例子中指定的起始行和结束行  
    public List<Test> find(String tableName, String startRow,  
            String stopRow) {  
        // 使用Scan可以实现过滤的功能  
        Scan scan = new Scan();  
        if (startRow == null) {  
            startRow = "";  
        }  
        if (stopRow == null) {  
            stopRow = "";  
        }  
        scan.setStartRow(Bytes.toBytes(startRow));  
        scan.setStopRow(Bytes.toBytes(stopRow));  
        return hbaseTemplate.find(tableName, scan, new RowMapper<Test>() {  
            public Test mapRow(Result result, int rowNum)  
                    throws Exception {  
                final Test info = new Test();  
                List<Cell> ceList = result.listCells();  
                if (ceList != null && ceList.size() > 0) {  
                    //Segment segment = new Segment();  
                    //Operator operator = new Operator();  
                    for (Cell cell : ceList) {  
                        if (null == info.getId() || info.getId().isEmpty()) {  
                            String row = Bytes.toString(cell.getRowArray(),cell.getRowOffset(), cell.getRowLength());  
                            info.setId(row);  
                        }  
                        // 获取值  
                        String value = Bytes.toString(cell.getValueArray(),cell.getValueOffset(), cell.getValueLength());  
                        // 获取列簇  
                        String family = Bytes.toString(cell.getFamilyArray(),cell.getFamilyOffset(), cell.getFamilyLength());  
                        // 获取列  
                        String quali = Bytes.toString(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength());  
                        info.setData(value);
                        // 以下为业务封装，可以忽略  
                        /*if (family.equals("operator")) {  
                            PropertiesCopyUtil.copy(operator, family, quali,value);  
                        } else {  
                            PropertiesCopyUtil.copy(segment, family, quali,value);  
                        } */ 
                        
                    }  
                }  
                return info;  
            }  
        });  
    }  
    // 获取一行的数据，并封装成vo  
    public Test get(String tableName, String rowName) {  
        return hbaseTemplate.get(tableName, rowName,  
                new RowMapper<Test>() {  
                    public Test mapRow(Result result, int rowNum)  
                            throws Exception {  
                        final Test info = new Test();  
                        List<Cell> ceList = result.listCells();  
                        if (ceList != null && ceList.size() > 0) {  
                            //Segment segment = new Segment();  
                            //Operator operator = new Operator();  
                            for (Cell cell : ceList) {  
                                if (null == info.getId() || info.getId().isEmpty()) {  
                                    String row = Bytes.toString(cell.getRowArray(),cell.getRowOffset(), cell.getRowLength());  
                                    info.setId(row);  
                                }  
                                // 获取值  
                                String value = Bytes.toString(cell.getValueArray(),cell.getValueOffset(), cell.getValueLength());  
                                // 获取列簇  
                                String family = Bytes.toString(cell.getFamilyArray(),cell.getFamilyOffset(), cell.getFamilyLength());  
                                // 获取列  
                                String quali = Bytes.toString(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength());  
                                /*if (family.equals("operator")) {  
                                    PropertiesCopyUtil.copy(operator, family, quali,value);  
                                } else {  
                                    PropertiesCopyUtil.copy(segment, family, quali,value);  
                                }  */
                                info.setData(value);
                            }  
                            //info.setOperator(operator);  
                            //info.setSegment(segment);  
                        }  
                        return info;  
                    }  
                });  
    }  
    // 将值插入表中指定的行，列簇，列中  
    public void put(final String tableName, final String rowKey,  
            final String familyName, final String column, final String value) {  
        hbaseTemplate.put(tableName, rowKey, familyName, column, value.getBytes());  
    }  
}
